package com.fyz.aaronfjava.util;

import com.alibaba.fastjson.JSON;
import com.fyz.aaronfjava.model.response.AsyncResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.util.CollectionUtils;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;

/**
 * @Author：AaronF
 * @Date：2024/1/31 10:16
 * @description 并发同步调用工具类 整合spring环境 如果不需要spring环境则更换为jdk线程池
 */
public class CountDownLatchUtil {

    private static final Logger log = LoggerFactory.getLogger(CountDownLatchUtil.class);

    @Autowired
    @Qualifier("asyncTaskExecutor")
    private AsyncTaskExecutor asyncTaskExecutor;

    /**
     * 开启异步线程执行任务 同步等待所有任务完成
     * @param tasks 任务集合 要求Callable接口返回值继承AsyncResponse
     * @return 结果集
     */
    public Map<String,AsyncResponse> async(Map<String, Callable<? extends AsyncResponse>> tasks){
        if (CollectionUtils.isEmpty(tasks)){
            return Collections.emptyMap();
        }
        // 根据任务数量创建同步器
        CountDownLatch countDownLatch = new CountDownLatch(tasks.size());
        Map<String,AsyncResponse> resultMap = new HashMap<>(tasks.size());
        // 并发调用
        tasks.forEach((key,callable) -> asyncTaskExecutor.execute(() -> {
            // 调用结果
            AsyncResponse result = new AsyncResponse();
            try {
                result = callable.call();
            } catch (Exception e) {
                result.setCode(AsyncResponse.FAILURE_CODE);
                result.setMsg(e.getMessage());
            }finally {
                // 保存调用结果
                log.info("保存异步调用结果{}:{}",key, JSON.toJSONString(result));
                resultMap.put(key,result);
                countDownLatch.countDown();
            }
        }));
        // 同步等待
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            log.warn("同步等待被中断");
            Thread.currentThread().interrupt();
        }
        return resultMap;
    }

    /**
     * 方便定义任务类型
     * @return
     */
    public Map<String, Callable<? extends AsyncResponse>> getTasks(){
        return new HashMap<>();
    }

}
